package com.heima.kafka.boot.stream;

import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;
import java.util.Arrays;

/*
定义为配置类
 */
@Configuration
public class StreamHandle {
    /**
     * 执行流式处理的方法，并交给ioc管理
     * @return
     */
    @Bean
    public KStream<String,String> kStream(StreamsBuilder streamsBuilder){
        //1.接收消息，此时此刻就是消费者的角色
        KStream<String, String> kStream = streamsBuilder.stream("stream-send-topic");
        //2.进行流式处理
        kStream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            /**
             *初始化操作
             * @param value  表示接收到的消息内容   "hello  itcast"   "hello kafka"
             * @return ["hello","itcast","hello","kafka"]
             */
            @Override
            public Iterable<String> apply(String value) {
                return Arrays.asList(value.split(" "));
            }
        })
                /**
                 * 表示分组，key=消息内容的key
                 * value=["hello","itcast","hello","kafka"]
                 */
                .groupBy((key,value)-> value)
                /**
                 * 事件窗口，可以处理晚到的数据，每10s处理一次
                 */
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //求和的api
                .count()
                .toStream()
                /**
                 * 到此，能够走到当前这步的话，表示流式处理完成
                 * 该map方法表示封装数据流式处理之后的数据，整理要发送给消费者的消息内容
                 */
                .map(new KeyValueMapper<Windowed<String>, Long, KeyValue<String, String>>() {
                    /**
                     * 表示处理消息，准备发送处理之后的消息给kafka
                     * @param key 表示要发送的消息，比如hello    itcast
                     * @param value 表示每个单词出现的次数
                     * @return
                     */
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, Long value) {
                        return new KeyValue<>(key.key(), value.toString());
                    }
                })
                /**
                 * 3.发送消息，指定消费者的topic，此时此刻作为生产者的角色
                 */
                .to("stream_get_topic");
            return kStream;
    }
}
